package com.airikeji.hr.rabbitmq;

import cn.hutool.core.lang.UUID;
import com.airikeji.hr.domain.MessageAlert;
import com.airikeji.hr.domain.MessageResend;
import com.airikeji.hr.service.MessageAlertService;
import com.airikeji.hr.service.MessageResendService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.MessageConversionException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.Optional;

/**
 * @Author huangrui
 * @Date 2021/7/8 17:38
 * @Version 1.0
 */
@Slf4j
@Service
@Component
public class RabbitMQSendUtil implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback{

    @Autowired
    private MessageResendService messageResendService;
    @Autowired
    private MessageAlertService messageAlertService;

    private RabbitTemplate rabbitTemplate;

    @Autowired
    public RabbitMQSendUtil(RabbitTemplate rabbitTemplate) {
        super();
        this.rabbitTemplate = rabbitTemplate;
        // 设置channel开启事务
        this.rabbitTemplate.setMandatory(true);
        this.rabbitTemplate.setReturnCallback(this);
        this.rabbitTemplate.setConfirmCallback(this);
    }

    /**
     * 从 String 转换为 byte[]
     * @param object
     * @return
     * @throws MessageConversionException
     */
    public Message toMessage(Object object) throws MessageConversionException {
        MessageProperties messageProperties = new MessageProperties();
        messageProperties.setContentType("text/plain");
        messageProperties.setMessageId(UUID.randomUUID().toString().replace("-",""));
        return new Message(object.toString().getBytes(), messageProperties);
    }

    /**
     * 从 String 转换为 byte[],已存在messageId时
     * @param object
     * @return
     * @throws MessageConversionException
     */
    public Message toMessage(Object object, String messageId) throws MessageConversionException {
        MessageProperties messageProperties = new MessageProperties();
        messageProperties.setContentType("text/plain");
        messageProperties.setMessageId(messageId);
        return new Message(object.toString().getBytes(), messageProperties);
    }


    /**
     * 从 byte[] 转换为 String
     * @param message
     * @return
     * @throws MessageConversionException
     */
    public Object fromMessage(Message message) throws MessageConversionException {
        String contentType = message.getMessageProperties().getContentType();
        if(null != contentType && contentType.contains("text")) {
            return new String(message.getBody());
        }
        return message.getBody();
    }

    /**
     * 发布消息（有路由key，如direct和topic模式）
     * @param
     */
    public void send(String exchange, String message, String routingKey) {
        Message messages = this.toMessage(message);
        //在fanoutExchange中在绑定Q到X上时，会自动把Q的名字当作bindingKey。
        try {
            this.rabbitTemplate.convertAndSend(exchange,routingKey,messages);
        }catch (Exception e){
            // 可以写保存到数据库后重发
            log.warn("发送消息至交换机失败，请联系管理员处理");
        }
    }

    /**
     * 发布消息（没有路由key，如fanout和Headers模式）
     * @param
     */
    public void send(String exchange, String message) {
        Message messages = this.toMessage(message);
        //在fanoutExchange中在绑定Q到X上时，会自动把Q的名字当作bindingKey。
        try {
            this.rabbitTemplate.convertAndSend(exchange,"",messages);
        }catch (Exception e){
            log.warn("发送消息至交换机失败，请联系管理员处理");
        }
    }


    /**
     * 用于消息重发的方法
     */
    @Transactional(rollbackFor = Exception.class)
    public void resend(MessageResend messageResend){
        try {
            // 如果重发次数大于3,则写入message_alert表,并且在重发表中删除该条消息
            if (messageResend.getResendNum() >= 3){
                MessageAlert messageAlert = new MessageAlert(messageResend.getId(),messageResend.getMessage(), messageResend.getRouteKey(), messageResend.getExchange());
                messageAlertService.insert(messageAlert);
                messageResendService.deleteById(messageResend.getId());
            }else {
                // 重发次数小于3,则继续重发
                Message message = this.toMessage(messageResend.getMessage(), messageResend.getId());
                String exchange = messageResend.getExchange();
                String routeKey = messageResend.getRouteKey();
                // 存在路由key
                if (messageResend.getRouteKey().length() != 0){
                    this.rabbitTemplate.convertAndSend(exchange,"", message);
                }else {
                    this.rabbitTemplate.convertAndSend(exchange, routeKey, message);
                }
            }
        }catch (Exception e){
            log.error("系统异常，请联系管理员处理!引起异常的原因可能是：发送消息至交换机失败、更新数据库message_resend或message_alert表异常！");
        }
    }

    // 消息确认
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        if (ack) {
            log.info("发送至交换机成功："+correlationData);
        } else {
            log.warn("发送至交换机失败，请保证交换机已存在或检查交换机名拼写正确"+correlationData+"；出现异常："+cause);
        }
    }

    //消息发送失败机制
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        log.info("执行消息重发机制！");
        String messageSting = this.fromMessage(message).toString(); // 需要重新发送的消息
        String messageId = message.getMessageProperties().getMessageId(); // 需要重新发送的消息的Id
        MessageResend messageResendInDataBase = messageResendService.queryById(messageId); // 从数据库中根据Id查询是否该记录
        if ((Optional.ofNullable(messageResendInDataBase).isPresent()) && (messageResendInDataBase.getResendNum() < 3)){
            // 如果messageResendInDataBase不为空,查询数据库中是否存在该消息,若消息存在该数据库且resendNum小于3，写入重发数据库message_resend数据库并给resendNum加1
            messageResendService.updateResendNum(messageId, messageResendInDataBase.getResendNum()+1);
            log.warn("消息发送失败-进入重发数据库，应答码：{}，原因：{}，交换机：{}，路由键：{}", replyCode, replyText, exchange, routingKey);
        }else {
            try {
                // 若该消息不存在，则插入该条数据
                messageResendService.insertResendMessage(new MessageResend(messageId, messageSting, routingKey, exchange, 1));
            }catch (Exception e){
                throw new RuntimeException("往message_resend表插入数据异常！可能由于id重复引起！");
            }
        }
    }
}

